package com.cn.remote;

import com.cn.coder.MessageDecoder;
import com.cn.coder.MessageEncoder;
import com.cn.common.*;
import com.cn.common.processor.INettyRequestProcessor;
import com.cn.exception.MyException;
import com.cn.msg.Message;
import com.cn.util.UtilAll;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;

import java.util.concurrent.*;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class RemoteClient2 extends AbStractNetty {

    private EventLoopGroup worker;
    private Bootstrap bootstrap;
    private boolean ok = false;
    private NettyClientConfig nettyClientConfig;
    private final ConcurrentMap<String, ChannelWrapper> channelTables = new ConcurrentHashMap<>();
    private final Lock channelLock = new ReentrantLock();


    public ExecutorService publicExecutorService = null;

    public RemoteClient2() {
        super(1000);
    }

    public RemoteClient2(NettyClientConfig nettyClientConfig) {
        super(1000);
        this.nettyClientConfig = nettyClientConfig;
    }

    public void start() {
        this.publicExecutorService = Executors.newFixedThreadPool(8);
        this.callbackExecutor = Executors.newFixedThreadPool(16);

        this.worker = new NioEventLoopGroup();
        this.bootstrap = new Bootstrap();
        this.bootstrap.group(worker);
        this.bootstrap.channel(NioSocketChannel.class);
        this.bootstrap.option(ChannelOption.TCP_NODELAY, true);
        this.bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000);
        this.bootstrap.handler(new RemoteClientHandler2());
    }

    public void shutdown() {
        if (this.worker != null) {
            this.worker.shutdownGracefully();
        }
    }

    public void registerExecutors(int code, INettyRequestProcessor nettyRequestProcessor, ExecutorService executorService) {
        if (executorService == null) {
            executorService = this.publicExecutorService;
        }
        Pair<INettyRequestProcessor, ExecutorService> pair = new Pair<>(nettyRequestProcessor, executorService);
        NettyRequestWrapper.add(code, pair);
    }

    /**
     * @param addrs 192.168.0.109:2101
     */
    public Channel createChannel(String addrs) {
        ChannelWrapper ch = this.channelTables.get(addrs);
        if (ch != null) {
            this.channelTables.remove(addrs);
            ch.getChannel().close();
        }

        //确保只有一个线程在创建初始化连接
        try {
            if (this.channelLock.tryLock(1000, TimeUnit.MILLISECONDS)) {
                try {
                    boolean createConnection = false;
                    ch = this.channelTables.get(addrs);
                    if (ch != null) {
                        if (ch.isOk()) {
                            ch.getChannel().close();
                            createConnection = true;
                            this.channelTables.remove(addrs);
                        } else if (!ch.getChannelFuture().isDone()) {
                            createConnection = false;
                        } else {
                            createConnection = true;
                        }
                    } else {
                        createConnection = true;
                    }

                    if (createConnection) {
                        ChannelFuture f = bootstrap.connect(RemoteHandler.string2SocketAddress(addrs));
                        ch = new ChannelWrapper(f);
                        this.channelTables.put(addrs, ch);
                    }
                } finally {
                    this.channelLock.unlock();
                }

            }
        } catch (InterruptedException e) {

        }

        if (ch != null) {
            ChannelFuture channelFuture = ch.getChannelFuture();
            if (channelFuture.awaitUninterruptibly(1000, TimeUnit.MILLISECONDS)) {
                return channelFuture.channel();
            } else {
                throw new MyException(String.format("%s 连接服务端超时，请确认服务端已经启动！", RemoteClient2.class.getName()));
            }
        }
        return null;
    }

    public Channel getChannel(String addrs) {
        ChannelWrapper ch = this.channelTables.get(addrs);
        if (ch != null) {
            if (ch.isOk()) {
                return ch.getChannel();
            }
        }
        return this.createChannel(addrs);
    }

    /**
     * 同步调用[有返回值]
     *
     * @param message
     * @param addr
     * @return
     */
    public byte[] invokeMsg(Message message, String addr) {
        return invokeMsg(message, addr, 30000);
    }

    public byte[] invokeMsg(Message message, String addr, long timeout) {
        //获取通道
        Channel channel = getChannel(addr);
        if (channel.isActive()) {
            message.setOpaque(UtilAll.getRequestId());
            try {
                ResponseFuture2 responseFuture2 = new ResponseFuture2(message.getOpaque());
                this.responseFutureTables.put(message.getOpaque(), responseFuture2);

                channel.writeAndFlush(message).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture channelFuture) throws Exception {
                        if (channelFuture.isSuccess()) {
                        } else {
                            responseFuture2.putResponse(null);
                        }
                    }
                });

                byte[] bytes = responseFuture2.await(timeout);//30秒后超时
                return bytes;
            } catch (Exception e) {
                e.printStackTrace();
                //移除通道
                this.channelTables.remove(addr);
            } finally {

            }

        } else {
            System.out.println("连接broker端失败");
            this.channelTables.remove(addr);
        }
        return null;
    }

    /**
     * 同步调用[无返回值]
     *
     * @param message
     * @param addr
     */
    public void invokeMsg2(Message message, String addr) {
        Channel channel = this.getChannel(addr);
        if (channel.isActive()) {
            channel.writeAndFlush(message).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if (channelFuture.isSuccess()) {
                    } else {
                    }
                }
            });

        }
    }

    /**
     * 异步调用[有返回值]
     *
     * @param message
     * @param addr
     * @param invokeCallback
     */
    public void invokeMsgAsync(Message message, String addr, InvokeCallback invokeCallback) {
        Channel channel = getChannel(addr);
        if (channel.isActive()) {
            try {
                boolean semaphore = this.semaphoreAsync.tryAcquire(3000, TimeUnit.MILLISECONDS);
                if (semaphore) {
                    message.setOpaque(UtilAll.getRequestId());
                    SeamphoreOneWayImpl oneWay = new SeamphoreOneWayImpl(semaphoreAsync);
                    ResponseFuture2 responseFuture2 = new ResponseFuture2(message.getOpaque(), invokeCallback, oneWay);
                    try {

                        this.responseFutureTables.put(message.getOpaque(), responseFuture2);

                        channel.writeAndFlush(message).addListener(new ChannelFutureListener() {
                            @Override
                            public void operationComplete(ChannelFuture channelFuture) throws Exception {
                                if (channelFuture.isSuccess()) {

                                } else {
                                    requireFail(message.getOpaque());
                                }

                            }
                        });
                    } catch (Exception e) {
                        e.printStackTrace();
                        //移除通道
                        this.channelTables.remove(addr);
                        //释放资源
                        responseFuture2.release();
                    }
                }
            } catch (InterruptedException e) {

                throw new MyException("请求太多，请稍后再试！");
            }

        } else {
            System.out.println("连接broker端失败");
            this.channelTables.remove(addr);
        }
    }

    public void requireFail(long opaque) {
        ResponseFuture2 responseFuture2 = this.responseFutureTables.get(opaque);
        responseFuture2.putResponse(null);
    }

    static class ChannelWrapper {
        private final ChannelFuture channel;

        public ChannelWrapper(ChannelFuture channel) {
            this.channel = channel;
        }

        public boolean isOk() {
            if (channel.channel() != null && channel.channel().isActive()) {
                return true;
            }
            return false;
        }

        public boolean isWriteAble() {
            return this.channel.channel().isWritable();
        }

        public Channel getChannel() {
            return this.channel.channel();
        }

        public ChannelFuture getChannelFuture() {
            return channel;
        }
    }

    class RemoteClientHandler2 extends ChannelInitializer<SocketChannel> {

        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            //如果消息是自定义类型，那么用以下处理器
            ch.pipeline().addLast(new MessageEncoder());//编码器
            ch.pipeline().addLast(new MessageDecoder());//解码器
            ch.pipeline().addLast(new IdleStateHandler(0, 0, 3, TimeUnit.SECONDS));//心跳设置
            ch.pipeline().addLast(new ClientConnectManageHandler2());//心跳处理器
            ch.pipeline().addLast(new ClientHandler2());//业务处理器
        }
    }

    /**
     * 心跳超时处理
     */
    class ClientConnectManageHandler2 extends ChannelDuplexHandler {

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if (evt instanceof IdleStateEvent) {
                IdleState state = ((IdleStateEvent) evt).state();
                if (state.equals(IdleState.ALL_IDLE)) {
                    Message message = new Message();
                    message.setType((byte) 1);//这里需要区分是 心跳包类型
                    message.setFlag((byte) 2);
                    message.setByteBodys("心跳".getBytes("UTF-8"));
                    ctx.writeAndFlush(message);
                }
            } else {
                super.userEventTriggered(ctx, evt);
            }
        }
    }

    /**
     * 自定义消息处理器
     */
    class ClientHandler2 extends ChannelInboundHandlerAdapter {


        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {

        }

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            process((Message) msg, ctx);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            System.out.println("与服务端断开连接了！");
//        System.out.println(cause.getMessage());
            cause.printStackTrace();
            ctx.close();
        }


    }
}
